package net.engio.mbassy.camel;

import net.engio.mbassy.listener.Listener;
import net.engio.mbassy.listener.Mode;

import org.apache.camel.AsyncCallback;
import org.apache.camel.AsyncProcessor;
import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.util.AsyncProcessorConverterHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Event bus listener, registered by Camel in MBassador.
 */
public class CamelEventListener {

    private static final transient Logger LOG = LoggerFactory.getLogger(CamelEventListener.class);

    /**
     * Unique ID of this listener object.
     */
    private final String listenerId;

    /**
     * MbassadorEndpoint object instance
     */
    private final MbassadorEndpoint eventBusEndpoint;

    /**
     * AsyncProcessor object instance which is used to process incoming messages asynchronously.
     */
    private final AsyncProcessor processor;

    /**
     * Class of events received from the Mbassador to be filtered.
     */
    private final Class<?> eventClass;

    /**
     * Creates a new instance of the CamelEventListener class.
     * @param listenerId unique ID of this listener object
     * @param eventBusEndpoint {@link MbassadorEndpoint} object instance
     * @param processor {@link Processor} object instance
     * @param eventClass {@link Class} of events received from the Mbassador to be filtered
     */
    public CamelEventListener(final String listenerId, final MbassadorEndpoint eventBusEndpoint,
                              final Processor processor, final Class<?> eventClass) {
        ObjectHelper.notNull(listenerId, "listenerId");
        ObjectHelper.notNull(eventBusEndpoint, "eventBusEndpoint");
        ObjectHelper.notNull(processor, "processor");

        this.listenerId = listenerId;
        this.eventBusEndpoint = eventBusEndpoint;
        this.processor = AsyncProcessorConverterHelper.convert(processor);
        this.eventClass = eventClass;
    }

    /**
     * Handles events received from the Mbassador.
     * Processes incoming messages asynchronously. 
     * @param event the event object to be processed
     */
    @Listener(delivery = Mode.Concurrent)
    public void handleEvent(final Object event) {
        LOG.trace("Received event: {}", event);
        if (eventClass == null || eventClass.isAssignableFrom(event.getClass())) {
            final Exchange exchange = eventBusEndpoint.createExchange(event);
            LOG.debug("Processing event: {}", event);
            // Will use async processor to support async routing engine
            processor.process(exchange, new AsyncCallback() {
                @Override
                public void done(final boolean doneSync) {
                    // nothing to do
                }
            });
        }
        else {
            if (LOG.isDebugEnabled()) {
                LOG.debug("Cannot process event: {} as its class type: {} is not assignable with: {}", new Object[] {
                        event, event.getClass().getName(), eventClass.getName() });
            }
        }
    }

    /**
     * Returns unique ID of this listener object.
     * @return unique ID of this listener object
     */
    public String getListenerId() {
        return listenerId;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + ((listenerId == null) ? 0 : listenerId.hashCode());
        result = prime * result + ((eventClass == null) ? 0 : eventClass.hashCode());
        return result;
    }

    @Override
    public boolean equals(final Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final CamelEventListener other = (CamelEventListener) obj;
        if (listenerId == null) {
            if (other.listenerId != null) {
                return false;
            }
        }
        else if (!listenerId.equals(other.listenerId)) {
            return false;
        }
        if (eventClass == null) {
            if (other.eventClass != null) {
                return false;
            }
        }
        else if (!eventClass.equals(other.eventClass)) {
            return false;
        }
        return true;
    }

}
